package com.bigweath.app.dwd.db;

import com.bigweath.utils.MyKafkaUtil;
import com.bigweath.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionComment {
    public static void main(String[] args) throws Exception {

        // TODO 1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 3. 从 KAFKA 读取业务数据，封装为 FLINK SQL表
        tableEnv.executeSql("create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string, string>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_interaction_comment"));

        // TODO 4.读取评论表数据
        Table commentInfo = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "data['order_id'] order_id, " +
                "data['create_time'] create_time, " +
                "data['appraise'] appraise, " +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'comment_info' " +
                "and `type` = 'insert' ");
        tableEnv.createTemporaryView("comment_info", commentInfo);

        // TODO 5.建立 MYSQL-LOOKUP 字典
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        // TODO 6.关联维度表
        Table resultTable = tableEnv.sqlQuery("select " +
                "ci.id, " +
                "ci.user_id, " +
                "ci.sku_id, " +
                "ci.order_id, " +
                "date_format(ci.create_time,'yyyy-MM-dd') date_id, " +
                "ci.create_time, " +
                "ci.appraise, " +
                "dic.dic_name, " +
                "ts " +
                "from comment_info ci " +
                "left join  " +
                "base_dic for system_time as of ci.proc_time as dic " +
                "on ci.appraise = dic.dic_code ");

        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 7.建立 Upsert-Kafka dwd_interaction_comment 表
        tableEnv.executeSql("create table dwd_interaction_comment(" +
                "id string, " +
                "user_id string, " +
                "sku_id string, " +
                "order_id string, " +
                "date_id string, " +
                "create_time string, " +
                "appraise string, " +
                "dic_name string, " +
                "ts string, " +
                "primary key(id) not enforced" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));

        // {"id":"1523202317074677761","user_id":"198","sku_id":"25","order_id":"26464","date_id":"2020-06-14","create_time":"2020-06-14 15:25:06","appraise":"1204","dic_name":"好评","ts":"1651994706"}
        tableEnv.executeSql("insert into dwd_interaction_comment select * from result_table").print();

        env.execute("DwdInteractionComment");


    }
}

